package cn.langpy.dblistener.core;


import cn.langpy.dblistener.core.model.DbType;
import cn.langpy.dblistener.core.model.OperateType;
import cn.langpy.dblistener.core.model.source.ColumnRecord;
import cn.langpy.dblistener.core.model.source.TableRecord;
import cn.langpy.dblistener.core.model.source.ColumnTypeRecord;
import cn.langpy.dblistener.core.model.transform.EventColumn;
import cn.langpy.dblistener.core.model.transform.EventRow;
import cn.langpy.dblistener.core.sqlparse.MysqlParser;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;

import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.Executor;
import java.util.logging.Logger;
import java.util.regex.Pattern;

/**
 * 事件解析
 * zhangchang
 */
public class ChangeConsumer implements DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>> {
    private static Logger log = Logger.getLogger(ChangeConsumer.class.toString());

    final static Map<String, OperateType> opMap = new HashMap<String, OperateType>() {
        {
            put("c", OperateType.INSERT);
            put("r", OperateType.READ);
            put("d", OperateType.DELETE);
            put("u", OperateType.UPDATE);
            put("create", OperateType.CREATE);
            put("drop", OperateType.DROP);
            put("truncate", OperateType.TRUNCATE);
            put("alter", OperateType.ALTER);
            put("UNKOWN", OperateType.ALTER);
        }
    };
    final static Map<DbType, SqlParser> sqlParserMap = new HashMap<DbType, SqlParser>() {
        {
            put(DbType.Mysql, new MysqlParser());
        }
    };
    private DbType type;
    private Executor executor;
    private Pattern excludedPattern;
    private List<String> tablesExclude = new ArrayList<>();
    private List<EventListener> listeners = new ArrayList<>();

    public ChangeConsumer(DbType type, List<EventListener> listeners, List<String> tablesExclude, Executor executor) {
        this.type = type;
        this.executor = executor;
        this.listeners.addAll(listeners);
        List<String> regexs = new ArrayList<>();
        if (tablesExclude!=null) {
            this.tablesExclude.addAll(tablesExclude);
            for (String s : tablesExclude) {
                regexs.add("(^" + s + ")");
            }
            excludedPattern = Pattern.compile(String.join("|", regexs));
        }

    }


    @Override
    public void handleBatch(List<ChangeEvent<String, String>> list, DebeziumEngine.RecordCommitter<ChangeEvent<String, String>> recordCommitter) throws InterruptedException {
        try {
            this.executor.execute(() -> {
                for (ChangeEvent<String, String> changeEvent : list) {
                    boolean needContinue = dealRecord(changeEvent);
                    if (!needContinue) {
                        break;
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public boolean dealRecord(ChangeEvent<String, String> changeEvent) {
        String value = changeEvent.value();
        if (value==null || "".equals(value) || "".equals(value.trim())) {
            return true;
        }
        JSONObject record = JSON.parseObject(value);
        JSONObject payload = record.getJSONObject("payload");
        JSONObject source = payload.getJSONObject("source");
        String database = source.getString("db");
        if (database == null) {
            return true;
        }
        String table = source.getString("table");
        if (table == null) {
            return true;
        }
        database = database.toLowerCase(Locale.ROOT);
        table = table.toLowerCase(Locale.ROOT);
        if (this.tablesExclude != null && this.tablesExclude.size() > 0) {
            if (isExcluded(database + "." + table)) {
                return true;
            }
        }
        JSONObject schema = record.getJSONObject("schema");
        String name = schema.getString("name");

        if (name.endsWith("SchemaChangeValue")) {
            /*ddl*/
            ddlParse(database, table, payload);
        } else {
            /*其他语句*/
            dataParse(database, table, payload, schema);
        }
        return true;
    }

    private void dataParse(String database, String table, JSONObject payload, JSONObject schema) {
        JSONArray schemaFields = schema.getJSONArray("fields");
        List<ColumnTypeRecord> beforeTypes = getTypes(schemaFields, "before");
        List<ColumnTypeRecord> afterTypes = getTypes(schemaFields, "after");
        String op = payload.getString("op");
        JSONObject jsonbefore = payload.getJSONObject("before");
        JSONObject jsonafter = payload.getJSONObject("after");
        List<EventColumn> bColumns = getEventColumns(jsonbefore, beforeTypes);
        List<EventColumn> aColumns = getEventColumns(jsonafter, afterTypes);
        EventRow eventRow = EventRow.builder()
                .ddl(false)
                .database(database)
                .table(table)
                .operate(opMap.get(op))
                .befores(bColumns)
                .afters(aColumns)
                .build();
        eventRow.setSql(sqlParserMap.get(this.type).getSql(database, table, eventRow));
        for (EventListener listener : listeners) {
            EventRow clone = null;
            try {
                clone = eventRow.clone();
            } catch (CloneNotSupportedException e) {
                e.printStackTrace();
                continue;
            }
            if (OperateType.DELETE == clone.getOperate()) {
                listener.onDelete(clone);
            } else if (OperateType.INSERT == clone.getOperate()) {
                listener.onInsert(clone);
            } else if (OperateType.READ == clone.getOperate()) {
                listener.onRead(clone);
            } else if (OperateType.UPDATE == clone.getOperate()) {
                listener.onUpdate(clone);
            }
        }

    }


    private List<EventColumn> getEventColumns(JSONObject json, List<ColumnTypeRecord> types) {
        List<EventColumn> columns = new ArrayList<EventColumn>();
        int size = types.size();
        for (int i = 0; i < size; i++) {
            ColumnTypeRecord typeRecord = types.get(i);
            EventColumn column = new EventColumn();
            column.setName(typeRecord.getField());
            column.setType(typeRecord.getType());
            if (json != null) {
                Object value = json.get(typeRecord.getField());
                if ("io.debezium.time.Date".equals(typeRecord.getName())) {
                    value = toDate(value);
                    column.setType("date");
                } else if ("io.debezium.time.Timestamp".equals(typeRecord.getName())) {
                    value = toDateTime(value);
                    column.setType("datetime");
                } else if ("io.debezium.data.Json".equals(typeRecord.getName())) {
                    column.setType("json");
                }
                if (column.getType().startsWith("int")) {
                    column.setComment("int");
                } else if (column.getType().startsWith("long")) {
                    column.setComment("long");
                } else if (column.getType().equals("datetime")) {
                    column.setComment("string");
                } else if (column.getType().equals("date")) {
                    column.setComment("string");
                } else if (column.getType().equals("float")) {
                    column.setComment("double");
                } else if (column.getType().equals("json")) {
                    column.setComment("json");
                } else {
                    column.setComment(column.getType());
                }
                column.setValue(value);
            }
            columns.add(column);
        }
        return columns;
    }

    private boolean isExcluded(String dbtable) {
        if (excludedPattern==null) {
            return false;
        }
        return excludedPattern.matcher(dbtable).find();
    }

    private String toDate(Object value) {
        if (value == null) {
            return null;
        }
        int valueLong = (int) value;
        LocalDate localDate = LocalDate.of(1970, 1, 1);
        localDate = localDate.plusDays(valueLong);
        return localDate.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
    }

    private String toDateTime(Object value) {
        if (value == null || value instanceof Integer) {
            return null;
        }
        long valueLong = (long) value;
        LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(valueLong), ZoneId.of("UTC"));
        return localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
    }

    private void ddlParse(String database, String table, JSONObject payload) {
        String ddl = (payload.getString("ddl") + "").toLowerCase(Locale.ROOT).trim();
        ddl = ddl.replace("`", "");
        String allTableName = (database + "." + table);
        if (!ddl.contains(allTableName)) {
            ddl = ddl.replace(table, allTableName);
        }
        int i1 = ddl.indexOf(" ");
        OperateType op = opMap.getOrDefault(ddl.substring(0, i1), OperateType.UNKOWN);
        if (ddl.startsWith("create") || ddl.startsWith("drop") || ddl.startsWith("alter") || ddl.startsWith("truncate")) {
            List<EventColumn> aColumns = new ArrayList<EventColumn>();
            JSONArray tableChanges = payload.getJSONArray("tableChanges");
            if (tableChanges != null && tableChanges.size() > 0) {
                JSONObject jsonObject = tableChanges.getJSONObject(0);
                if (OperateType.CREATE.equals(op)) {
                    ddl = ddl.replace("table ", "table if not exists ");
                    JSONObject jsonObject1 = jsonObject.getJSONObject("table");
                    TableRecord tableRecord = JSON.to(TableRecord.class,jsonObject1);
                    List<ColumnRecord> columns = tableRecord.getColumns();
                    int size = columns.size();
                    for (int i = 0; i < size; i++) {
                        ColumnRecord columnRecord = columns.get(i);
                        String type = columnRecord.getTypeName().toLowerCase(Locale.ROOT);
                        EventColumn column = new EventColumn();
                        column.setName(columnRecord.getName());
                        if ("timestamp".equals(type)) {
                            type = "datetime";
                        }
                        column.setType(type);
                        column.setLength(columnRecord.getLength());
                        column.setScale(columnRecord.getScale());
                        column.setComment(columnRecord.getComment());
                        column.setNullable(columnRecord.getOptional());
                        if (tableRecord.getPrimaryKeyColumnNames().contains(column.getName())) {
                            column.setPrimary(true);
                        } else {
                            column.setPrimary(false);
                        }
                        aColumns.add(column);
                    }
                }
            }
            EventRow eventRow = EventRow.builder()
                    .ddl(true)
                    .database(database)
                    .table(table)
                    .operate(op)
                    .sql(ddl)
                    .afters(aColumns)
                    .build();
            for (EventListener listener : listeners) {
                EventRow clone = null;
                try {
                    clone = eventRow.clone();
                } catch (CloneNotSupportedException e) {
                    e.printStackTrace();
                    continue;
                }
                if (OperateType.CREATE == clone.getOperate()) {
                    listener.onCreate(clone);
                } else if (OperateType.ALTER == clone.getOperate()) {
                    listener.onAlter(clone);
                } else if (OperateType.TRUNCATE == clone.getOperate()) {
                    listener.onTruncate(clone);
                } else if (OperateType.DROP == clone.getOperate()) {
                    listener.onDrop(clone);
                }
            }
        }
    }

    private List<ColumnTypeRecord> getTypes(JSONArray schemaFields, String scope) {
        List<ColumnTypeRecord> types = null;
        for (Object schemaField : schemaFields) {
            JSONObject jsonField = (JSONObject) schemaField;
            String field = jsonField.getString("field");
            if (scope.equals(field)) {
                JSONArray fields = jsonField.getJSONArray("fields");
                types = fields.toList(ColumnTypeRecord.class);
                break;
            }
        }
        return types;
    }

}
